"""会话与消息管理路由"""
import json
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from loguru import logger

from backend.common.config import config_manager
from backend.database.db import db_manager
from backend.deps.auth import get_current_user


router = APIRouter(
    prefix=f"/{config_manager.api.version}",
    tags=["sessions"],
)

AUTH_ENABLED = getattr(config_manager.auth, "feature_auth", True)
ANONYMOUS_USER_ID = "anonymous"
UNOWNED_SESSION_VALUES = {None, "", "default", "anonymous"}


class SessionResponse(BaseModel):
    """会话信息响应模型"""

    id: str
    name: str
    startTime: datetime
    lastMessageTime: datetime
    messageCount: int = 0


class SessionListResponse(BaseModel):
    """会话列表响应"""

    sessions: List[SessionResponse]
    total: int


class MessageResponse(BaseModel):
    """历史消息响应模型"""

    id: str
    session_id: str
    role: str
    content: str
    created_at: datetime
    metadata: Dict[str, Any] = Field(default_factory=dict)
    rag_context: Optional[List[Dict[str, Any]]] = None


class MessageListResponse(BaseModel):
    """历史消息列表响应"""

    items: List[MessageResponse]
    total: int


class CreateSessionRequest(BaseModel):
    """创建会话请求体"""

    id: Optional[str] = None
    title: Optional[str] = None
    name: Optional[str] = None


class UpdateSessionRequest(BaseModel):
    """更新会话标题请求体"""

    title: Optional[str] = Field(None, min_length=1)
    name: Optional[str] = Field(None, min_length=1)


class CreateMessageRequest(BaseModel):
    """创建消息请求体"""

    role: str
    content: str
    metadata: Optional[Dict[str, Any]] = None
    rag_context: Optional[List[Dict[str, Any]]] = None
    created_at: Optional[datetime] = None


class BatchCreateMessagesRequest(BaseModel):
    """批量写入消息请求体，用于历史导入/迁移，不参与实时聊天流程。"""

    messages: List[CreateMessageRequest]
    purpose: str = Field(
        default="import",
        description="用途说明，仅用于审计，常见值: import, restore",
    )


class BatchDeleteSessionsRequest(BaseModel):
    """批量删除会话请求体"""

    session_ids: List[str] = Field(..., min_items=1)


class ExportSessionsRequest(BaseModel):
    """导出会话请求体"""

    session_ids: Optional[List[str]] = None


class ExportSessionsResponse(BaseModel):
    """导出会话响应"""

    sessions: List[Dict[str, Any]]
    total: int


def _parse_datetime(value: Any) -> datetime:
    """将任意时间值转换为datetime对象"""
    if isinstance(value, datetime):
        return value
    if isinstance(value, str):
        for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S"):
            try:
                return datetime.strptime(value, fmt)
            except ValueError:
                continue
    return datetime.utcnow()


def _row_to_session(row: Any) -> SessionResponse:
    """统一转换数据库会话行"""
    if isinstance(row, dict):
        created_at = _parse_datetime(row.get("created_at"))
        updated_at = _parse_datetime(row.get("updated_at") or created_at)
        message_count = int(row.get("message_count") or 0)
        title = row.get("title") or row.get("name") or "未命名对话"
        return SessionResponse(
            id=str(row.get("id")),
            name=title,
            startTime=created_at,
            lastMessageTime=updated_at,
            messageCount=message_count,
        )

    id_value = str(row[0])
    title_value = row[1] or "未命名对话"
    created_at = _parse_datetime(row[2])
    updated_at = _parse_datetime(row[3]) if row[3] else created_at
    message_count = int(row[4] or 0)
    return SessionResponse(
        id=id_value,
        name=title_value,
        startTime=created_at,
        lastMessageTime=updated_at,
        messageCount=message_count,
    )


def _normalize_message_row(row: Any) -> MessageResponse:
    """统一转换数据库消息行"""
    if isinstance(row, dict):
        metadata = row.get("metadata") or row.get("meta") or {}
        if isinstance(metadata, str):
            try:
                metadata = json.loads(metadata)
            except Exception:
                metadata = {}
        rag_context = None
        if isinstance(metadata, dict) and "rag_context" in metadata:
            rag_context = metadata.pop("rag_context")
        return MessageResponse(
            id=str(row.get("id")),
            session_id=str(row.get("session_id")),
            role=row.get("role"),
            content=row.get("content"),
            created_at=_parse_datetime(row.get("created_at")),
            metadata=metadata or {},
            rag_context=rag_context,
        )

    metadata = {}
    if len(row) > 5 and row[5]:
        try:
            metadata = json.loads(row[5]) if isinstance(row[5], str) else row[5]
        except Exception:
            metadata = {}
    rag_context = None
    if isinstance(metadata, dict) and "rag_context" in metadata:
        rag_context = metadata.pop("rag_context")
    return MessageResponse(
        id=str(row[0]),
        session_id=str(row[1]),
        role=row[2],
        content=row[3],
        created_at=_parse_datetime(row[4]),
        metadata=metadata or {},
        rag_context=rag_context,
    )


def _effective_user_id(current_user: dict) -> str:
    """获取实际使用的用户ID"""
    return current_user.get("id") if AUTH_ENABLED else ANONYMOUS_USER_ID


def _ensure_session_access(session_id: str, current_user: dict) -> Dict[str, Any]:
    """确保当前用户有权访问目标会话"""
    session = db_manager.get_session_by_id(session_id)
    if not session:
        raise HTTPException(status_code=404, detail="会话不存在")

    if AUTH_ENABLED:
        user_id = current_user.get("id")
        owner = session.get("user_id")
        if owner in UNOWNED_SESSION_VALUES:
            if db_manager.claim_session_owner(session_id, user_id):
                session = db_manager.get_session_by_id(session_id) or session
                owner = session.get("user_id")
        if owner not in UNOWNED_SESSION_VALUES and owner != user_id:
            raise HTTPException(status_code=403, detail="无权访问该会话")
        session["user_id"] = user_id
    else:
        session["user_id"] = ANONYMOUS_USER_ID
    return session


def _delete_session_rows(session_id: str) -> Dict[str, int]:
    """删除会话及其关联数据"""
    deleted_messages = db_manager.delete_session_messages(session_id)
    with db_manager.get_cursor() as cursor:
        if db_manager.db_type == "sqlite":
            cursor.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
        else:
            cursor.execute("DELETE FROM sessions WHERE id = %s", (session_id,))
    try:
        db_manager.delete_session_model(session_id)
    except Exception as exc:
        logger.debug(
            "删除会话模型映射失败，但不影响主流程",
            session_id=session_id,
            error=str(exc),
        )
    return {"deleted_messages": deleted_messages}


def _build_placeholders(count: int) -> str:
    """根据数据库类型构造IN子句占位符"""
    placeholder = "%s" if db_manager.db_type != "sqlite" else "?"
    return ",".join([placeholder] * count)


def _extract_count(row: Any) -> int:
    """统一解析COUNT结果"""
    if row is None:
        return 0
    if isinstance(row, dict):
        return int(row.get("count") or row.get("count(*)") or 0)
    if isinstance(row, (list, tuple)):
        return int(row[0]) if row else 0
    return int(row)


@router.get("/sessions", response_model=SessionListResponse)
async def list_sessions(
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
    search: Optional[str] = Query(None, description="会话标题关键字"),
    current_user: dict = Depends(get_current_user),
):
    """获取当前用户的会话列表"""
    user_id = _effective_user_id(current_user)
    placeholder = "%s" if db_manager.db_type != "sqlite" else "?"
    conditions: List[str] = []
    params: List[Any] = []

    if AUTH_ENABLED:
        conditions.append(f"s.user_id = {placeholder}")
        params.append(user_id)
    if search:
        conditions.append(f"s.title LIKE {placeholder}")
        params.append(f"%{search.strip()}%")

    where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""

    with db_manager.get_cursor(dict_cursor=db_manager.db_type != "sqlite") as cursor:
        cursor.execute(f"SELECT COUNT(*) FROM sessions s {where_clause}", tuple(params))
        total = _extract_count(cursor.fetchone())

        data_params = list(params)
        data_params.extend([limit, offset])
        cursor.execute(
            f"""
            SELECT s.id, s.title, s.created_at, s.updated_at,
                   COALESCE(cnt.message_count, 0) AS message_count
            FROM sessions s
            LEFT JOIN (
                SELECT session_id, COUNT(*) AS message_count
                FROM messages
                GROUP BY session_id
            ) cnt ON s.id = cnt.session_id
            {where_clause}
            ORDER BY s.updated_at DESC, s.created_at DESC
            LIMIT {placeholder} OFFSET {placeholder}
            """,
            tuple(data_params),
        )
        rows = cursor.fetchall()

    sessions = [_row_to_session(row) for row in rows]
    return SessionListResponse(sessions=sessions, total=total)


@router.post("/sessions", response_model=SessionResponse)
async def create_session(
    request: CreateSessionRequest,
    current_user: dict = Depends(get_current_user),
):
    """创建新会话"""
    session_id = (request.id or "").strip() or str(uuid.uuid4())
    raw_title = (request.title or request.name or "").strip() or "未命名对话"
    now = datetime.utcnow()
    user_id = _effective_user_id(current_user)

    existing = db_manager.get_session_by_id(session_id)
    if existing:
        if AUTH_ENABLED:
            owner = existing.get("user_id")
            if owner not in UNOWNED_SESSION_VALUES and owner != user_id:
                raise HTTPException(status_code=403, detail="会话ID已被占用")
        existing["title"] = existing.get("title") or raw_title
        existing["message_count"] = db_manager.count_session_messages(session_id)
        return _row_to_session(existing)

    with db_manager.get_cursor() as cursor:
        if db_manager.db_type == "sqlite":
            cursor.execute(
                """
                INSERT INTO sessions (id, user_id, title, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                (session_id, user_id, raw_title, now, now),
            )
        else:
            cursor.execute(
                """
                INSERT INTO sessions (id, user_id, title, created_at, updated_at)
                VALUES (%s, %s, %s, %s, %s)
                """,
                (session_id, user_id, raw_title, now, now),
            )

    return SessionResponse(
        id=session_id,
        name=raw_title,
        startTime=now,
        lastMessageTime=now,
        messageCount=0,
    )


@router.get("/sessions/{session_id}", response_model=SessionResponse)
async def get_session_detail(session_id: str, current_user: dict = Depends(get_current_user)):
    """导出会话及其消息"""
    session = _ensure_session_access(session_id, current_user)
    session["message_count"] = db_manager.count_session_messages(session_id)
    return _row_to_session(session)


@router.patch("/sessions/{session_id}", response_model=SessionResponse)
async def update_session(
    session_id: str,
    request: UpdateSessionRequest,
    current_user: dict = Depends(get_current_user),
):
    """更新会话标题"""
    new_title = (request.title or request.name or "").strip()
    if not new_title:
        raise HTTPException(status_code=400, detail="会话标题不能为空")

    session = _ensure_session_access(session_id, current_user)
    now = datetime.utcnow()
    with db_manager.get_cursor() as cursor:
        if db_manager.db_type == "sqlite":
            cursor.execute(
                "UPDATE sessions SET title = ?, updated_at = ? WHERE id = ?",
                (new_title, now, session_id),
            )
        else:
            cursor.execute(
                "UPDATE sessions SET title = %s, updated_at = %s WHERE id = %s",
                (new_title, now, session_id),
            )

    session["title"] = new_title
    session["updated_at"] = now
    session["message_count"] = db_manager.count_session_messages(session_id)
    return _row_to_session(session)


@router.delete("/sessions/{session_id}")
async def delete_session(session_id: str, current_user: dict = Depends(get_current_user)):
    """删除单个会话"""
    _ensure_session_access(session_id, current_user)
    stats = _delete_session_rows(session_id)
    return {"success": True, "deleted": 1, **stats}


@router.delete("/sessions/batch")
async def delete_sessions_batch(
    request: BatchDeleteSessionsRequest,
    current_user: dict = Depends(get_current_user),
):
    """批量删除会话"""
    deleted = 0
    for session_id in request.session_ids:
        try:
            _ensure_session_access(session_id, current_user)
        except HTTPException as exc:
            if exc.status_code in {403, 404}:
                continue
            raise
        _delete_session_rows(session_id)
        deleted += 1
    return {"success": True, "deleted": deleted}


@router.get("/sessions/{session_id}/messages", response_model=MessageListResponse)
async def get_session_messages(
    session_id: str,
    limit: int = Query(200, ge=1, le=500),
    offset: int = Query(0, ge=0),
    current_user: dict = Depends(get_current_user),
):
    """获取指定会话的历史消息列表"""
    _ensure_session_access(session_id, current_user)
    raw_items = db_manager.get_session_messages(session_id, limit=limit, offset=offset)
    items = [_normalize_message_row(item) for item in raw_items]
    total = db_manager.count_session_messages(session_id)
    return MessageListResponse(items=items, total=total)


@router.post("/sessions/{session_id}/messages", response_model=MessageResponse)
async def create_session_message(
    session_id: str,
    request: CreateMessageRequest,
    current_user: dict = Depends(get_current_user),
):
    """写入单条消息"""
    session = _ensure_session_access(session_id, current_user)
    payload_created_at = request.created_at or datetime.utcnow()
    metadata = request.metadata or {}
    rag_context = request.rag_context or None
    message_id = db_manager.insert_session_message(
        session_id=session_id,
        role=request.role,
        content=request.content,
        created_at=payload_created_at,
        metadata=metadata,
        rag_context=rag_context,
        user_id=session.get("user_id"),
    )
    return MessageResponse(
        id=str(message_id or uuid.uuid4()),
        session_id=session_id,
        role=request.role,
        content=request.content,
        created_at=payload_created_at,
        metadata=metadata,
        rag_context=rag_context,
    )


@router.post("/sessions/{session_id}/messages/batch")
async def create_session_messages_batch(
    session_id: str,
    request: BatchCreateMessagesRequest,
    current_user: dict = Depends(get_current_user),
):
    """批量写入消息，仅用于历史导入或恢复备份."""
    if not request.messages:
        return {"success": True, "inserted": 0}

    session = _ensure_session_access(session_id, current_user)
    inserted = 0
    for message in request.messages:
        try:
            payload_created_at = message.created_at or datetime.utcnow()
            db_manager.insert_session_message(
                session_id=session_id,
                role=message.role,
                content=message.content,
                created_at=payload_created_at,
                metadata=message.metadata or {},
                rag_context=message.rag_context or None,
                user_id=session.get("user_id"),
            )
            inserted += 1
        except Exception as exc:
            logger.warning("写入单条消息失败", session_id=session_id, error=str(exc), role=message.role)
            continue
    return {"success": True, "inserted": inserted, "purpose": request.purpose}


@router.post("/sessions/export", response_model=ExportSessionsResponse)
async def export_sessions(
    request: ExportSessionsRequest,
    current_user: dict = Depends(get_current_user),
):
    """导出会话及其消息"""
    session_ids = request.session_ids
    exported: List[Dict[str, Any]] = []

    with db_manager.get_cursor(dict_cursor=db_manager.db_type != "sqlite") as cursor:
        placeholder = "%s" if db_manager.db_type != "sqlite" else "?"
        params: List[Any] = []
        conditions: List[str] = []
        if AUTH_ENABLED:
            conditions.append(f"user_id = {placeholder}")
            params.append(_effective_user_id(current_user))
        if session_ids:
            placeholders = _build_placeholders(len(session_ids))
            conditions.append(f"id IN ({placeholders})")
            params.extend(session_ids)
        where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""
        cursor.execute(
            f"SELECT id, title, created_at, updated_at FROM sessions {where_clause}",
            tuple(params),
        )
        sessions = cursor.fetchall()

    for row in sessions:
        session_id = row[0] if isinstance(row, (list, tuple)) else row.get("id")
        try:
            messages = db_manager.get_session_messages(session_id, limit=1000, offset=0)
            exported.append(
                {
                    "session": _row_to_session(row).dict(),
                    "messages": [_normalize_message_row(msg).dict() for msg in messages],
                }
            )
        except Exception as exc:
            logger.warning("导出会话失败", session_id=session_id, error=str(exc))
            continue

    return ExportSessionsResponse(sessions=exported, total=len(exported))
